package com.gitee.watermaker


import java.util.UUID

import com.gitee.watermaker.OrderWaterMarkDemo.{BoundedOutOfOrdernessGenerator, Order}
import org.apache.commons.lang3.time.FastDateFormat
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.source.{RichSourceFunction, SourceFunction}
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala.function.WindowFunction
import org.apache.flink.streaming.api.scala.{DataStream, OutputTag, StreamExecutionEnvironment}
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

import scala.collection.mutable.ArrayBuffer
import scala.util.Random

object OrderWaterMarkFull {

  //订单样例类Order(订单ID、用户ID、订单金额、时间戳(事件时间))
  case class Order(orderId: String, userId: Int, money: Long, timestamp: Long, systems: Long)

  //定义一个时间转换工具
  val df: FastDateFormat = FastDateFormat.getInstance("HH:mm:ss")

  def main(args: Array[String]): Unit = {
    val senv: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    //设置事件时间
    senv.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    //设置水印间隔,默认200毫秒
    senv.getConfig.setAutoWatermarkInterval(200)

    //准备乱序的订单数据
    import org.apache.flink.api.scala._
    val orderDS: DataStream[Order] = senv.addSource(new RichSourceFunction[Order] {
      var isRunning = true

      override def run(ctx: SourceFunction.SourceContext[Order]): Unit = {
        while (isRunning) {
          val orderId: String = UUID.randomUUID().toString.replaceAll("-", "")
          val userId: Int = Random.nextInt(1)
          val money: Int = Random.nextInt(101)
          val timestamp: Long = System.currentTimeMillis() - Random.nextInt(20) * 1000 //减去一个随机的x秒时间表示该订单是该x秒之前产生的
          val systems: Long = System.currentTimeMillis() //系统时间
          val order = Order(orderId, userId, money, timestamp, systems)
          ctx.collect(order)
          Thread.sleep(1000)
        }
      }

      override def cancel(): Unit = {
        isRunning = false
      }
    })

    //使用事件时间分配水印
    val waterMarksData: DataStream[Order] = orderDS.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[Order](Time.milliseconds(3000)) {
      override def extractTimestamp(element: Order): Long = {
        //从element中获取事件时间
        val eventTime: Long = element.timestamp
        println(s"数据的事件时间:${df.format(eventTime)},数据被分配的水印时间:${df.format(eventTime - 3 * 1000)},当前系统时间:${df.format(element.systems)}")
        eventTime
      }
    })

    //准备一个OutPutTag用来存放超时的数据
    val lateDataArr = new OutputTag[Order]("lateData")

    val windowDetail: DataStream[String] = waterMarksData.keyBy(_.userId)
      .timeWindow(Time.seconds(5))
      .allowedLateness(Time.seconds(3)) //超时的时间
      .sideOutputLateData(lateDataArr) //存放的位置
      .apply(new WindowFunction[Order, String, Int, TimeWindow] {
      override def apply(key: Int, window: TimeWindow, input: Iterable[Order], out: Collector[String]): Unit = {
        //将属于每一个窗口中的所有数据事件时间,水印时间,窗口开始和结束组成一个字符串并返回
        val eventTimeArr = ArrayBuffer[String]() //准备存放eventTime的可变数组
        val watermakertimeArr = ArrayBuffer[String]() //准备存放watermaker的可变数组
        input.foreach(order => {
          val eventTime: String = df.format(order.timestamp)
          val watermakertime: String = df.format(order.timestamp - 3000)
          eventTimeArr.append(eventTime)
          watermakertimeArr.append(watermakertime)
        })
        //获取窗口开始和结束
        val start: String = df.format(window.getStart)
        val end: String = df.format(window.getEnd)
        val str = s"key:${key} ,窗口开始:${start},窗口结束:${end},属于该窗口的数据的事件时间:${eventTimeArr},水印时间:${watermakertimeArr}"
        out.collect(str)
      }
    })
    windowDetail.print()

    //打印出超过wm的数据
    windowDetail.getSideOutput(lateDataArr).print("迟到的数据:>>>>>>>>>>>>>>>")

    senv.execute()
  }
}
